#!/usr/bin/env python3
"""The call agent's one feed, run under a Monitor: every call the operator is on, as it is said, for as long as
the agent's session lives. The agent is started before any call and stays through all of them; this is what it
is parked on between calls and what it hears during them.

  aux-listen --id <short session id> [--batch 2s]

Aux's call watcher (CallWatcher.swift) writes $AUX_HOME/call.json ({"env","call","at"}) once a call has
connected and removes it when the operator leaves. While it names a call, this runs `tuple capture follow
<call>` and passes its lines on; between calls it waits. A sidebar, a rejoin and the next call are all just the
next thing call.json names.

Lines printed (each one an event for the agent; the call's own lines always begin with a timestamp):
  [aux] call env=<env> call=<id> cli=<path> | run: <aux-boot … --call <id>>     the operator is on this call
  [aux] call over call=<id>      they left it: nothing more for that call
  [aux] rules changed            me.md changed (the operator told Aux how to behave): re-read its Standing constraints
  [aux] rearm                    25 minutes passed (a Monitor lasts 30): start this command again
  [aux] superseded               another session is the call agent now: stop this one
  [aux] the Tuple CLI says: …    an error from the CLI, once

Across a re-arm nothing is lost or repeated: the follow replays the call so far, and the lines already
delivered ($AUX_HOME/listen.json remembers the last few hundred, and the time of the newest) are dropped. Every line is also logged with when it arrived
($AUX_HOME/calls/<call>/feed.log, for aux-latency), and agent.json carries a heartbeat (`listeningAt`), the
current call and when the last line came (`lastCallAt`), which is how the app knows the agent is alive.
"""
import argparse
import os
import re
import select
import signal
import subprocess
import sys
import time
import zlib
from datetime import datetime
from pathlib import Path

# Inside an app bundle a __pycache__ write is macOS's "python3 was prevented from modifying apps" prompt.
sys.dont_write_bytecode = True
sys.path.insert(0, str(Path(__file__).resolve().parent))
import aux_context as ctx  # noqa: E402

BOOT = Path(__file__).resolve().parent / "aux-boot"
STATE = ctx.AUX / "listen.json"
PIDFILE = ctx.AUX / "listen.pid"
REARM_AFTER = float(os.environ.get("AUX_LISTEN_REARM") or 25 * 60)
BEAT_EVERY = 15
STAMP = re.compile(r"^\[(\d[^\]]*)\]")   # a call's line starts with its time; anything else is the CLI talking
REMEMBER = 600            # delivered lines remembered per call, to tell a replayed line from a new one
OLD = 120                 # a replayed line this many seconds older than the newest delivered was delivered too
# Seconds the stream's own "call ended" is believed while Aux still names the call.
OVER_HOLD = float(os.environ.get("AUX_LISTEN_OVER_HOLD") or 20)
NOISE = re.compile(r"^\[[^\]]*\] · transcription_started ")


def say(line):
    sys.stdout.write(line + "\n")
    sys.stdout.flush()


def take_over():
    """One listener per folder, the newest: an earlier instance (a Monitor started twice) is ended."""
    try:
        old = int(PIDFILE.read_text().strip())
        if old != os.getpid():
            os.kill(old, signal.SIGTERM)
    except (OSError, ValueError):
        pass
    PIDFILE.write_text(f"{os.getpid()}\n")


class Follow:
    """`tuple capture follow <call>` for one call, read without blocking."""

    def __init__(self, cli, call, batch):
        cmd = [cli, "capture", "follow", call] + (["--batch", batch] if batch else [])
        self.proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL)
        self.fd = self.proc.stdout.fileno()
        os.set_blocking(self.fd, False)
        self.buffer = b""
        self.eof = False

    def lines(self, timeout):
        """The complete lines that arrive within `timeout` seconds."""
        if self.eof:
            time.sleep(timeout)
            return []
        ready, _, _ = select.select([self.fd], [], [], timeout)
        if not ready:
            return []
        try:
            chunk = os.read(self.fd, 65536)
        except BlockingIOError:
            return []
        if not chunk:
            self.eof = True
            rest, self.buffer = self.buffer, b""
            return [rest.decode("utf-8", "replace")] if rest.strip() else []
        self.buffer += chunk
        *whole, self.buffer = self.buffer.split(b"\n")
        return [w.decode("utf-8", "replace") for w in whole]

    def stop(self):
        if self.proc.poll() is None:
            self.proc.terminate()
            try:
                self.proc.wait(timeout=3)
            except subprocess.TimeoutExpired:
                self.proc.kill()


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--id", required=True, help="this session's short id, as agent.json has it")
    ap.add_argument("--batch", default="", help="pass --batch to the follow (only after the Monitor suppressed output)")
    a = ap.parse_args()
    signal.signal(signal.SIGTERM, lambda *_: sys.exit(0))
    if ctx.agent_record().get("id") != a.id:
        say("[aux] superseded")
        return 3
    take_over()

    state = ctx.read_json(STATE)
    if state.get("id") != a.id:
        state = {}   # another session's: this one has heard nothing yet
    # Per call: {"newest": the latest timestamp delivered, "seen": checksums of the last lines delivered}.
    calls = state.get("calls") if isinstance(state.get("calls"), dict) else {}
    announced = state.get("announced") or []
    heard = int(ctx.agent_record().get("heard") or 0)
    started = time.time()
    last_beat = 0.0
    last_line_at = 0.0
    dirty = False
    rules_at = mtime(ctx.WIKI / "me.md")
    follow = None        # the running follow
    on = None            # (env, call, at): the call being followed, or just left by a follow that will restart
    over = {}            # (call, at) the stream itself said had ended, and when: not followed again for OVER_HOLD
    told_over = set()    # (call, at) the agent has been told is over
    retry_at, backoff = 0.0, 1.0
    last_error = ""
    feed = None

    def save():
        nonlocal dirty
        keep = dict(sorted(calls.items(), key=lambda kv: kv[1].get("newest", ""))[-5:])
        ctx.write_json(STATE, {"id": a.id, "calls": keep, "announced": announced})
        dirty = False

    def beat(extra=None):
        nonlocal last_beat
        last_beat = time.time()
        fields = {"state": "ready", "listeningAt": int(last_beat), "heard": heard}
        if last_line_at:
            fields["lastCallAt"] = int(last_line_at)
        fields.update(extra or {})
        return ctx.update_agent(fields, only_id=a.id)

    def call_over(call, at):
        nonlocal last_line_at
        if (call, at) in told_over:
            return
        told_over.add((call, at))
        say(f"[aux] call over call={call}")
        ctx.status("done", call=call)
        last_line_at = time.time()

    def announce(env, call, at, cli):
        nonlocal announced
        announced = [call, at]
        told_over.discard((call, at))
        save()
        ctx.status("catching_up", call=call)
        say(f"[aux] call env={env} call={call} cli={cli} | run: AUX_HOME='{ctx.AUX}' '{BOOT}' --env {env} --call {call}")

    def is_new(call, line):
        """False for a line the follow is replaying. Marks a new one delivered."""
        m = STAMP.match(line)
        if not m:
            return True
        rec = calls.setdefault(call, {"newest": "", "seen": []})
        mark = zlib.crc32(line.encode())
        if mark in rec["seen"]:
            return False
        if rec["newest"] and seconds(rec["newest"]) - seconds(m.group(1)) > OLD:
            return False
        rec["seen"] = (rec["seen"] + [mark])[-REMEMBER:]
        rec["newest"] = max(rec["newest"], m.group(1))
        return True

    def close():
        nonlocal follow, feed
        if follow:
            follow.stop()
            follow = None
        if feed:
            feed.close()
            feed = None
        if dirty:
            save()

    beat()
    try:
        while True:
            now = time.time()
            if now - last_beat >= BEAT_EVERY:
                if beat().get("id") != a.id:
                    say("[aux] superseded")
                    return 3
                if dirty:
                    save()
            if now - started > REARM_AFTER:
                say("[aux] rearm")
                return 0

            live = ctx.read_json(ctx.LIVE_CALL)
            cur = (live.get("env"), live.get("call"), live.get("at")) if live.get("call") and live.get("env") in ctx.CLIS else None
            if on and cur != on:
                # The operator is no longer on it (or is on another: a sidebar).
                was = on
                close()
                on = None
                call_over(was[1], was[2])
            held = cur and now - over.get((cur[1], cur[2]), 0) < OVER_HOLD
            if cur and not follow and not held and now >= retry_at:
                env, call, at = cur
                cli = ctx.cli_path(env) or ctx.CLIS[env]
                (ctx.AUX / "calls" / call).mkdir(parents=True, exist_ok=True)   # also what names this folder as the call's
                try:
                    follow = Follow(cli, call, a.batch)
                except OSError as e:
                    follow = None
                    retry_at, backoff = now + backoff, min(backoff * 2, 15)
                    if str(e) != last_error:
                        last_error = str(e)
                        say(f"[aux] can't follow the call: {e}")
                    continue
                on = cur
                feed = open(ctx.AUX / "calls" / call / "feed.log", "a", buffering=1)
                beat({"env": env, "call": call})
                if announced != [call, at]:
                    announce(env, call, at, cli)

            if not follow:
                time.sleep(0.5)
                continue

            env, call, at = on
            for line in follow.lines(1.0):
                if not line.strip() or NOISE.match(line):
                    continue
                if not STAMP.match(line):
                    # The CLI talking, not the call. Its end marker stops the follow being restarted for a while;
                    # anything else (an error) is passed on once, however often the follow is retried.
                    if "call_ended" in line:
                        over[(call, at)] = time.time()
                    elif "stream_opened" not in line and "caught_up" not in line and line != last_error:
                        last_error = line
                        say(f"[aux] the Tuple CLI says: {line}")
                    continue
                if not is_new(call, line):
                    continue   # delivered before the re-arm (or a restart of the follow)
                dirty = True
                if (call, at) in told_over:
                    # The stream said the call had ended, yet here is more of it: the operator is back on it.
                    over.pop((call, at), None)
                    announce(env, call, at, ctx.cli_path(env) or ctx.CLIS[env])
                if mtime(ctx.WIKI / "me.md") != rules_at:
                    rules_at = mtime(ctx.WIKI / "me.md")
                    say("[aux] rules changed")
                say(line)
                heard += len(line)
                last_line_at = time.time()
                backoff = 1.0
                try:
                    feed.write(f"{time.time():.3f}\t{line}\n")
                except OSError:
                    pass
            if follow.eof:
                # The stream ended: the operator left the call (the marker says so), or transcription restarted.
                close()
                if now - over.get((call, at), 0) < OVER_HOLD + 5:
                    call_over(call, at)
                else:
                    retry_at, backoff = time.time() + backoff, min(backoff * 2, 15)
    finally:
        close()
        try:
            if PIDFILE.read_text().strip() == str(os.getpid()):
                PIDFILE.unlink()
        except OSError:
            pass


def seconds(stamp):
    try:
        return datetime.fromisoformat(stamp.replace("Z", "+00:00")).timestamp()
    except ValueError:
        return 0.0


def mtime(path):
    try:
        return path.stat().st_mtime
    except OSError:
        return 0.0


if __name__ == "__main__":
    sys.exit(main())
